package com.clw;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.clw.bean.Sku;
import com.clw.utils.DimUtil;
import com.clw.utils.JdbcUtil;
import com.clw.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

/**
 * 利用旁路缓存进行数据关联
 */
public class SideCache {

    public static void main(String[] args) throws Exception {

        // 创建流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 获取kafka 的流
        FlinkKafkaConsumer<String> consumer = MyKafkaUtil.getKafkaConsumer("wide_table", "dssd");
        //
        DataStreamSource<String> orderWideStream = env.addSource(consumer);

        SingleOutputStreamOperator<JSONObject> process = orderWideStream.map(x -> {
            return JSON.parseObject(x);
        }).process(new ProcessFunction<JSONObject, JSONObject>() {
                       @Override
                       public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, JSONObject>.Context context,
                                                  Collector<JSONObject> collector) throws Exception {

                           String sku_id = jsonObject.getString("sku_id");
                           // 根据sku id 获取其他数据
                           JSONObject skuInfo = DimUtil.getDimInfo(JdbcUtil.getConnect(), "sku", "sku_id", sku_id, Sku.class);
                           jsonObject.put("sku_name",skuInfo==null?"没有找到sku名称":skuInfo.getString("sku_name"));
                           collector.collect(jsonObject);
                       }
                   }

        );
        process.print();

        env.execute();


    }

}
